//package main.scala.com.hu.wc
//
//import java.text.SimpleDateFormat
//import java.util
//import java.util.{Calendar, Collections, Date}
//
//import org.apache.flink.api.common.ExecutionConfig
//import org.apache.flink.api.common.typeutils.TypeSerializer
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
//import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner
//import org.apache.flink.streaming.api.windowing.triggers.{EventTimeTrigger, Trigger}
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow
///**
// * @Author: hujianjun
// * @Date: 2020/12/23 7:37
// * @Describe:
// */
////输入数据
//case class Top100Input(event_id: String, date_d: String, timeStamp: Long, uid: Long, weekTag: String, monthTag: String)
//
//object CustomWindow {
//  def main(args: Array[String]): Unit = {
//
//  }
//}
//
//class CustomWindowAssigner[T](tag: String) extends WindowAssigner[T, TimeWindow] {
//  //窗口分配的主要方法，需要为每一个元素指定所属的分区
//  override def assignWindows(element: T, timestamp: Long, context: WindowAssigner.WindowAssignerContext): util.Collection[TimeWindow] = {
//    var offset: (Long, Long) = null
//    tag match {
//      case "month" => offset = getTimestampFromMon(timestamp)
//      case "week" => offset = getTimestampFromWeek(timestamp)
//    }
//    //分配窗口
//    Collections.singletonList(new TimeWindow(offset._1, offset._2))
//  }
//
//  //注意此处需要进行类型的转换，否则或编译出错，java版本好像没问题，但是java对于上面的offset处理有点难搞，所以放弃了
//  override def getDefaultTrigger(env: StreamExecutionEnvironment): Trigger[T, TimeWindow] = EventTimeTrigger.create().asInstanceOf[Trigger[T, TimeWindow]]
//
//  override def getWindowSerializer(executionConfig: ExecutionConfig): TypeSerializer[TimeWindow] = new TimeWindow.Serializer
//
//  //是否使用事件时间
//  override def isEventTime: Boolean = true
//
//
//  /**
//   * 获取指定时间戳当月时间戳范围
//   * eg:2020-03-12 11:35:13 (timestamp=1583984113960l)
//   * 结果为：(1582992000000,1585670399999)=>(2020-03-01 00:00:00,2020-03-31 23:59:59)
//   *
//   * @param timestamp 时间戳
//   * @return
//   */
//  def getTimestampFromMon(timestamp: Long): (Long, Long) = {
//    val calendar = Calendar.getInstance()
//    calendar.setTime(DateHelper.getInstance().getDateFromStr(new SimpleDateFormat("yyyyMM01000000").format(new Date(timestamp)), "yyyyMMddHHmmss"))
//    val numsOfMon: Long = calendar.getActualMaximum(Calendar.DAY_OF_MONTH)
//    calendar.set(Calendar.DAY_OF_MONTH, 1)
//    val start: Long = calendar.getTimeInMillis
//    val end: Long = start + numsOfMon * 24 * 60 * 60 * 1000 - 1
//    (start, end)
//  }
//
//  /**
//   * 获取指定时间戳本周时间范围（从周日开始）
//   * eg:2020-03-14 23:59:59 (timestamp=1583895064000l)
//   * 结果为：(1583596800000,1584201599999)=>(2020-03-08 00:00:00,2020-03-14 23:59:59)
//   *
//   * @param timestamp 时间戳
//   * @return
//   */
//  def getTimestampFromWeek(timestamp: Long): (Long, Long) = {
//    val calendar = Calendar.getInstance()
//    calendar.setTime(DateHelper.getInstance().getDateFromStr(new SimpleDateFormat("yyyyMMdd000000").format(new Date(timestamp)), "yyyyMMddHHmmss"))
//    //    calendar.setFirstDayOfWeek(Calendar.SUNDAY)//设置周日为首日  默认值，一般不用设置
//    calendar.set(Calendar.DAY_OF_WEEK, Calendar.SUNDAY)
//    val start: Long = calendar.getTimeInMillis
//    (start, start + 7 * 24 * 60 * 60 * 1000l - 1)
//  }
//}
